package com.doitedu.core

import com.doitedu.utils.SparkUtil
import org.apache.spark.rdd.RDD

/**
 * @Date: 22.7.2 
 * @Author: HANGGE
 * @qq: 598196583
 * @Tips: 学大数据 ,到多易教育
 * @Description:
 */
object C18_SortByKey {
  def main(args: Array[String]): Unit = {

    val sc = SparkUtil.getSparkContext("sortByKey")

    /**
     * 1,zss1,M,北京,23
     * 2,zss2,M,北京,21
     * 3,zss4,M,南京,24
     * 4,zss3,M,北京,24
     * 5,zss5,F,南京,27
     * 6,zss6,M,北京,23
     */
    val dataRDD = sc.textFile("data/log/")
    val tpRDD = dataRDD.map(line => {
      val arr = line.split(",")
      (arr(0).toInt, arr(1), arr(2), arr(3), arr(4).toInt)
    })
   // testSortBy(tpRDD)
  // 将数据封装成KV类型
    val tpRDD2= dataRDD.map(line => {
      val arr = line.split(",")
      ( arr(4).toInt , (arr(0).toInt, arr(1), arr(2), arr(3), arr(4).toInt))
    })
    // 按照key排序
    val sortedByKeyRDD = tpRDD2.sortByKey()
   // repartition算子
    val resRDD = sortedByKeyRDD.repartition(1)
    // 观察现象
    resRDD.map(tp=>tp._2).foreach(println)
    sc.stop

  }

  private def testSortBy(tpRDD: RDD[(Int, String, String, String, Int)]): Unit = {
    // 先在各个分区内排序
    val sortedRDD = tpRDD.sortBy(_._1)
    // 将排好序的数据输出磁盘
    //  sortedRDD.saveAsTextFile("data/output/")
    // 设置成一个分区  所有数据在一个分区中  数据全局有序
    val rdd = sortedRDD.coalesce(1)
    rdd.foreach(println)
    // 将结果数据收集到本地
    //  sortedRDD.collect().foreach(println)
  }
}
